package cn.ilikes.tools.nosql.jpa.clients.dataSources.hbases;

import java.io.IOException;
import java.util.Properties;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTablePool;
import org.slf4j.Logger;

import cn.ilikes.tools.logs.Loggerfactory;
import cn.ilikes.tools.nosql.jpa.JPASQLPersistenUnitEnv;
import cn.ilikes.tools.nosql.jpa.JPASQLRunTimeException;
import cn.ilikes.tools.nosql.jpa.SysConfigs;
import cn.ilikes.tools.nosql.jpa.clients.dataSources.AbDataSource;
import cn.ilikes.tools.nosql.jpa.clients.dataSources.JPAHBaseDataSourceRunTimeException;
import cn.ilikes.tools.nosql.jpa.clients.dataSources.JPASQLDataSourceException;
import cn.ilikes.tools.nosql.jpa.configs.databases.DataStoreConfigs;

/**
 * HBASE Data Source
 * 
 * @author fack
 * 
 */
public class HbaseDataSource extends AbDataSource<HTablePool> {

	private Logger logger = Loggerfactory.instance(HbaseDataSource.class);

	private enum State {
		init, opens, close,
	}

	private State state = State.close;

	/** The Constant DEFAULT_POOL_SIZE. */
	private static final int DEFAULT_POOL_SIZE = 100;

	private static String hbase_zookeeper_quorum = "hbase.zookeeper.quorum";

	private static String hbase_auto_flush = "false";

	private static String hbase_zookeeper_property_client_port = "hbase.zookeeper.property.clientPort";

	private HTablePool hTablePool;

	public HbaseDataSource(JPASQLPersistenUnitEnv persistenUnitEnv) {
		super(persistenUnitEnv);

	}

	@Override
	public void close() {
		try {
			if (!validateOpen())return;
			hTablePool.close();
			state = State.close;
			Loggerfactory.info(logger, getPersistenUnitEnv().getPersistenUnitName() + " 's " + getPersistenUnitEnv().getPersistenceUnitMetadata().getDataStore().getName()
					+ " datastore   success   close !  ");
		} catch (IOException e) {
			Loggerfactory.error(logger, getPersistenUnitEnv().getPersistenUnitName() + " 's " + getPersistenUnitEnv().getPersistenceUnitMetadata().getDataStore().getName()
					+ " datastore no   success close !  ", e);

		}

	}

	@Override
	public void open() {
		try {
			if (validateOpen()) {
				Loggerfactory.info(logger, getPersistenUnitEnv().getPersistenUnitName() + " 's " + getPersistenUnitEnv().getPersistenceUnitMetadata().getDataStore().getName()
						+ " datastore   opened !  ");
				return;
			}
			Loggerfactory.info(logger, "Start open " + getPersistenUnitEnv().getPersistenUnitName() + " 's "
					+ getPersistenUnitEnv().getPersistenceUnitMetadata().getDataStore().getName() + " datastore !  ");
			DataStoreConfigs dataStoreConfigs = getPersistenUnitEnv().getPersistenceUnitMetadata().getDataStore();
			HTablePool tablePool = createTablePool(dataStoreConfigs);
			this.sethTablePool(tablePool);
			state = State.opens;
			Loggerfactory.info(logger, getPersistenUnitEnv().getPersistenUnitName() + " 's " + getPersistenUnitEnv().getPersistenceUnitMetadata().getDataStore().getName()
					+ "   datastore open  success  !  ");
		} catch (JPASQLDataSourceException e) {
			Loggerfactory.error(logger, getPersistenUnitEnv().getPersistenUnitName() + " 's " + getPersistenUnitEnv().getPersistenceUnitMetadata().getDataStore().getName()
					+ " start  datastore no open  success  !  ", e);

		}

	}

	private void sethTablePool(HTablePool hTablePool) {
		this.hTablePool = hTablePool;
	}

	@Override
	public void init() throws JPASQLDataSourceException {
		Loggerfactory.info(logger, "Star Init " + getPersistenUnitEnv().getPersistenUnitName() + "  Persisten  Unit  nme  "
				+ getPersistenUnitEnv().getPersistenceUnitMetadata().getDataStore().getName() + " 'Hbase  DataStore");
		DataStoreConfigs dataStoreConfigs = getPersistenUnitEnv().getPersistenceUnitMetadata().getDataStore();
		validateZookeeper(dataStoreConfigs);
		validateMaster(dataStoreConfigs);
		state = State.init;

	}

	private void validateMaster(DataStoreConfigs dataStoreConfigs) throws JPASQLDataSourceException {
		Loggerfactory.info(logger, getPersistenUnitEnv().getPersistenUnitName() + "  Persisten  Unit 's " + getPersistenUnitEnv().getPersistenceUnitMetadata().getDataStore().getName()
				+ " 'DataStore validate  Master  ");
		Properties properties = dataStoreConfigs.getConnection().getProperties();
		String node = properties.getProperty(SysConfigs.hbase_datasore_master_node);
		String port = properties.getProperty(SysConfigs.hbase_datasore_master_port);
		String poolSize = properties.getProperty(SysConfigs.hbase_datasore_master_pool_size);

		if (node == null)
			throw new JPASQLDataSourceException("dataStore  properties  " + SysConfigs.hbase_datasore_master_node + " not set value ");

		if (port == null)
			throw new JPASQLDataSourceException("dataStore  properties  " + SysConfigs.hbase_datasore_master_port + " not set value ");

		if (poolSize == null)
			throw new JPASQLDataSourceException("dataStore  properties  " + SysConfigs.hbase_datasore_master_pool_size + " not set value ");

	}

	private void validateZookeeper(DataStoreConfigs dataStore) throws JPASQLDataSourceException {
		Loggerfactory.info(logger, getPersistenUnitEnv().getPersistenUnitName() + "  Persisten  Unit 's " + getPersistenUnitEnv().getPersistenceUnitMetadata().getDataStore().getName()
				+ " 'DataStore validate  Zookeeper ");
		Properties properties = dataStore.getConnection().getProperties();

		String zookeeperHost = properties.getProperty(SysConfigs.hbase_datasore_zookeeper_quorum);

		String zookeeperPort = properties.getProperty(SysConfigs.hbase_datasore_zookeeper_client_port);

		if (zookeeperHost == null)
			throw new JPASQLDataSourceException("dataStore  properties  " + SysConfigs.hbase_datasore_zookeeper_quorum + " not set value ");

		if (zookeeperPort == null)
			throw new JPASQLDataSourceException("dataStore  properties  " + SysConfigs.hbase_datasore_zookeeper_client_port + " not set value ");
	}

	private HTablePool createTablePool(DataStoreConfigs dataStore) throws JPASQLDataSourceException {
		if (hTablePool == null) {
			synchronized (this) {
				if (dataStore == null)
					throw new JPASQLDataSourceException("dataStore is null");
				try {
					Loggerfactory.info(logger, "start  create " + dataStore.getName() + " hbase table Pool ! ");
					Properties properties = dataStore.getConnection().getProperties();
					String node = properties.getProperty(SysConfigs.hbase_datasore_master_node);
					String port = properties.getProperty(SysConfigs.hbase_datasore_master_port);
					String poolSize = properties.getProperty(SysConfigs.hbase_datasore_master_pool_size);
					Loggerfactory.info(logger, "hbase table Pool config   node : " + node + " port : " + port + "  poolSize : " + poolSize);
					Configuration hadoopConf = new Configuration();
					hadoopConf.set("hbase.master", node + ":" + port);
					String zookeeperHost = properties.getProperty(SysConfigs.hbase_datasore_zookeeper_quorum);
					String zookeeperPort = properties.getProperty(SysConfigs.hbase_datasore_zookeeper_client_port);
					Loggerfactory.info(logger, "hbase zookeeper  host : " + zookeeperHost + " port : " + zookeeperPort);
					hadoopConf.set(hbase_zookeeper_quorum, zookeeperHost != null ? zookeeperHost : node);
					hadoopConf.set(hbase_zookeeper_property_client_port, zookeeperPort != null ? zookeeperPort : "2181");
					int poolSize_ = poolSize != null ? Integer.parseInt(poolSize) : DEFAULT_POOL_SIZE;
					HTablePool hTablePool = new HTablePool(hadoopConf, poolSize_, Boolean.valueOf(hbase_auto_flush));
					return hTablePool;
				} catch (Exception e) {
					throw new JPASQLRunTimeException(e);
				}
			}
		} else {
			return this.hTablePool;
		}

	}

	@Override
	public boolean isOpen() {
		return state.equals(State.opens);
	}

	@Override
	protected HTablePool createDataSource() {
		validateState();
		return this.hTablePool;
	}

	private void validateState() {
		if (!this.state.equals(State.opens))
			throw new JPAHBaseDataSourceRunTimeException("this  data source no  open ! ");
	}

	@Override
	public boolean validateOpen() {
		return state.equals(State.opens);
	}

}